package com_.dxy_test_.zookeeper_.distributed_lock_;

import com_.dxy_test_.zookeeper_.ZkConfig;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.StringUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 该版本为zk分布式锁的升级版本，正确规避惊群效应
 *
 *  简单的方案：
 * 加锁就设置个临时节点即可，解锁就删除，并且其他线程都监听这个节点，加入500个线程再等待
 *
 * 解锁时，这500个线程个都会监听到已解锁，都会竞争锁，这样500个线程一起动作，就会引起服务震荡，这就是“惊群效应”
 *
 * 就好像一头狼进入了羊群
 *
 * 高级方案：
 *
 * 采用临时有序节点，顺序最小的那个就获得锁，同时监听前一个节点
 *
 * 按节点顺序获得锁，没有一堆线程都一下子监听到了
 *
 * 参考：
 * Zookeeper分布式锁遇到惊群效应如何解决：
 * https://blog.csdn.net/aazhzhu/article/details/89967346
 *
 *
 * @Author: dangxianyue
 * @Date: 2020/8/25 18:10
 */
public class MyZkLock implements Lock {

    /**
     * 锁的名称，父节点名，持久化节点，/lockName/000001, /lockName/000002
     */
    private String lockName;

    /**
     * zk 客户端，对zk进行操作完成锁操作
     */
    private ZkClient client;

    /**
     * 当前锁的节点，使用ThreadLocal存储，可以把锁作为实例变量或类变量
     * 若currentPath写做实例变量，则锁只能作为局部变量，每个线程都需要创建一个
     */
    private ThreadLocal<String> currentPath = new ThreadLocal<>();

    /**
     * 前一个节点，ThreadLocal作用和上面相同
     */
    private ThreadLocal<String> beforePath = new ThreadLocal<>();

    /**
     * 重入锁计数器
     */
    private AtomicInteger reentryLockCount = new AtomicInteger(0);

    /**
     * 初始化参数
     *
     * @param lockName
     */
    public MyZkLock(String lockName) {
        if (StringUtils.isEmpty(lockName)) {
            throw new IllegalArgumentException("锁名称不能为空");
        }
        if (!lockName.startsWith("/")) lockName = "/" + lockName;
        this.lockName = lockName;
        // 创建客户端对象
        client = ZkConfig.zkClient;
        // 创建持久节点
        if (!client.exists(this.lockName)) {
            client.createPersistent(this.lockName);
        }
    }

    /**
     * 使用锁的入口
     */
    @Override
    public void lock() {
        if (!tryLock()) {
            // 等待释放锁
            waitForLock();
            // 递归调用自己
            lock();
        }
    }

    /**
     * 尝试获取锁
     *
     * 重点逻辑：
     * 判断当前线程创建的节点是否是顺序最小的节点，如果是则获得锁
     * 如果不是则监控它的前一个节点，这样有效地避免了惊群效应
     *
     * 但这样只能实现的是公平锁，非公平锁就是所有线程监听一个临时节点，但会有惊群效应，引起服务震荡
     *
     * @return
     */
    @Override
    public boolean tryLock() {
        // 当前节点为空则创建临时顺序节点
        if (this.currentPath.get() == null || !client.exists(this.currentPath.get())) {
            String node = this.client.createEphemeralSequential(lockName + "/", "1");
            currentPath.set(node);
        }
        // 获取当前节点下所有的子节点
        List<String> children = client.getChildren(lockName);
        // 对节点进行由小到大排序
        Collections.sort(children);
        String firstChild = lockName + "/" + children.get(0);
        // 如果当前节点是最小的节点则获取到锁
        if (firstChild.equals(currentPath.get())) {
            reentryLockCount.getAndIncrement();
            return true;
        } else {
            // 得到字节的索引号（获取当前节点的索引序号）
            int curIndex = children.indexOf(currentPath.get().substring(lockName.length() + 1));
            // 获取前一个索引节点
            String beforeNode = lockName + "/" + children.get(curIndex - 1);
            beforePath.set(beforeNode);
            return false;
        }
    }

    /**
     * 释放锁的逻辑
     */
    @Override
    public void unlock() {
        int count = reentryLockCount.get();
        if (this.currentPath != null && count == 1) {
            client.delete(currentPath.get());
            currentPath.set(null);
            reentryLockCount.getAndDecrement();
        }else {
            reentryLockCount.getAndDecrement();
        }

    }

    /**
     * 等待释放锁
     */
    private void waitForLock() {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // zk数据节点的监听器
        IZkDataListener dataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) {
            }
            @Override
            public void handleDataDeleted(String dataPath) {
                countDownLatch.countDown();
            }
        };
        // 绑定监听节点
        client.subscribeDataChanges(beforePath.get(), dataListener);
        if (this.client.exists(beforePath.get())) {
            try {
                //前一个节点存在，线程会阻塞在这里，知道前一节点解锁被删除了，上面会监听到，countdown=0，被唤醒
                //然后再重新调用tryLock逻辑
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 解除监听
        client.unsubscribeDataChanges(beforePath.get(), dataListener);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) {
        return false;
    }


    @Override
    public void lockInterruptibly() throws InterruptedException {

    }


    @Override
    public Condition newCondition() {
        return null;
    }
}
